package com.myflink.day03;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

/**
 * @author Shelly An
 * @create 2020/9/18 14:31
 */
public class Sink_MySQL {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<String> fileDS = env.readTextFile("input/sensor-data.log");

        fileDS.addSink(new RichSinkFunction<String>() {
            //内存泄露 指的是 垃圾太多了，gc还来不及做就已经泄露了
            //内存溢出 指的是 运行中的数据太多了，导致数据溢出
            private Connection conn = null;
            private PreparedStatement pstmt = null;


            @Override
            public void open(Configuration parameters) throws Exception {
                System.out.println("连接数据库");
                conn = DriverManager.getConnection(
                        "jdbc:mysql://localhost:3306/myemployees",
                        "root",
                        "root");
                pstmt = conn.prepareStatement("INSERT INTO sensor values (?,?,?)");
            }

            //会发现关闭两次，是因为fileDS是有界数据流
            @Override
            public void close() throws Exception {
                pstmt.close();
                conn.close();
                System.out.println("关闭数据库");
            }

            //Deprecated是指不建议用，不是说以后会取消
            @Override
            public void invoke(String value, Context context) throws Exception {
                String[] datas = value.split(",");
                System.out.println(value);
                pstmt.setString(1, datas[0]);
                pstmt.setLong(2, Long.parseLong(datas[1]));
                pstmt.setInt(3, Integer.parseInt(datas[2]));

                pstmt.execute();
            }
        });

        env.execute();
    }
}
